package com.coding4play.mqtt.broker.service;

import com.coding4play.mqtt.broker.internal.InternalMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;


@Component
@Slf4j
public class KafkaService {

//    @Inject
//    private KafkaProducer kafkaProducer;
//    @Inject
//    private BrokerProperties brokerProperties;

    @Async
    public void send(InternalMessage internalMessage) {
        log.info("send to kafka todo");
//        try {
//            //消息体转换为Hex字符串进行转发
//            ProducerRecord<String, String> data = new ProducerRecord<>(brokerProperties.getProducerTopic(), internalMessage.getTopic(), JSONObject.toJSONString(internalMessage));
//            kafkaProducer.send(data,
//                    new Callback() {
//                        public void onCompletion(RecordMetadata metadata, Exception e) {
//                            if (e != null) {
//                                e.printStackTrace();
//                                log.error(e.getMessage(), e);
//                            } else {
//                                log.info("The offset of the record we just sent is: " + metadata.offset());
//                            }
//                        }
//                    });
//        } catch (Exception e) {
//            log.error("kafka没有连接成功..");
//        }
    }

}
